package cn.z2huo.demo.kafka.apache.configtest.consumer.fetch;

import cn.z2huo.demo.kafka.apache.consumer.ConsumerProperties;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.RandomStringUtils;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.TopicPartition;

import java.time.Duration;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Properties;

/**
 * <p>fetch.max.wait.ms 和 fetch.min.bytes 配置测试
 *
 *
 * @author z2huo
 */
@Slf4j
public class FetchBytesWaitConfigConsumer {

    private static final String TOPIC = "test-topic";

    private static final String GROUP_ID = "my-consumer-group-1";

    private static final int COMMIT_BUFFER_SIZE = 10;

    public static void main(String[] args) {
        new FetchBytesWaitConfigConsumer().consume();
    }

    /*
     * <p>查看下面的两行日志，间隔 10 秒，而配置的两个参数分别为 10000000 bytes 和 10000 ms
     *
     * [19:31:20.139][INFO ] cn.z2huo.demo.kafka.apache.configtest.consumer.fetch.FetchBytesWaitConfigConsumer:55 consume - Received message, partition is 0, offset is 40, key is partition-0, value is adsfads
     * [19:31:30.524][INFO ] cn.z2huo.demo.kafka.apache.configtest.consumer.fetch.FetchBytesWaitConfigConsumer:55 consume - Received message, partition is 1, offset is 34, key is null, value is {"id":53015298592047319,"userCode":"userCode viJnzglx","userName":"userName PMihorXl","validDate":"2025-04-27","invalidDate":"2026-04-27","validFlag":"1","deleteFlag":"0","createTime":"2025-04-27T19:29:35.1669804","operateTime":"2025-04-27T19:29:35.1669804"}
     */

    /**
     * <p>手动提交偏移量
     * <p>手动提交也需要指定 group.id，否则会报错 To use the group management or offset commit APIs, you must provide a valid group.id in the consumer configuration.
     */
    public void consume() {
        Properties properties = ConsumerProperties.getConsumerProperties();
        properties.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
        properties.setProperty(ConsumerConfig.GROUP_ID_CONFIG, GROUP_ID + "-" + RandomStringUtils.randomNumeric(4));
        // 设置为 30 万字节，因为生成了 1000 条 user 数据，一个 user 数据 json 字符串字节为 256 字节左右，1000 个，25 万字节
        properties.setProperty(ConsumerConfig.FETCH_MIN_BYTES_CONFIG, "3000000");
//        properties.setProperty(ConsumerConfig.FETCH_MAX_WAIT_MS_CONFIG, "10000");
        properties.setProperty(ConsumerConfig.FETCH_MAX_WAIT_MS_CONFIG, "5000");

        try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(properties)) {
            consumer.subscribe(Collections.singletonList(TOPIC));
            List<ConsumerRecord<String, String>> bufferList = new ArrayList<>(10);
            while (true)  {
//                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
//                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(10000));
                // poll 方法入参中的时间间隔 和 ConsumerConfig.FETCH_MAX_WAIT_MS_CONFIG 取最小值
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
                if (records.isEmpty()) {
                    log.info("poll records is empty");
                } else {
                    log.info("poll records size is {}, contains partition is {}", records.count(),
                            records.partitions().stream().map(TopicPartition::partition).toList());
                }
                for (ConsumerRecord<String, String> record : records) {
                    bufferList.add(record);
                }
                if (bufferList.size() >= COMMIT_BUFFER_SIZE) {
                    log.info("bufferList size is {}, bigger than {}, start process", bufferList.size(), COMMIT_BUFFER_SIZE);
                    log.info("bufferList start partition is {}, offset is {}, end partition is {} offset is {}", bufferList.getFirst().partition(), bufferList.getFirst().offset(),
                            bufferList.getLast().partition(), bufferList.getLast().offset());
                    for (ConsumerRecord<String, String> record : bufferList) {
                        log.info("Received message, partition is {}, offset is {}, key is {}, value is {}",
                                record.partition(), record.offset(), record.key(), record.value());
                    }
                    consumer.commitSync();
                    bufferList.clear();
                }
            }
        }
    }

}
